package com.hillky.proxy.handler;

import com.hillky.IdGenerator;
import com.hillky.NettyBootstrapInitializer;
import com.hillky.YrpcBootstrap;
import com.hillky.annotation.TryTimes;
import com.hillky.compress.CompressorFactory;
import com.hillky.discovery.Registry;
import com.hillky.enumeration.RequestType;
import com.hillky.enumeration.RespCode;
import com.hillky.exceptions.DiscoveryException;
import com.hillky.exceptions.NetworkException;
import com.hillky.protection.CircuitBreaker;
import com.hillky.protection.RateLimiter;
import com.hillky.serialize.SerializerFactory;
import com.hillky.transport.message.RequestPayload;
import com.hillky.transport.message.YrpcRequest;
import com.hillky.transport.message.YrpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.annotation.Annotation;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程嗲用过程都封装在了invoke方法中
 * 1、发现可用服务
 * 2、建立连接
 * 3、发送请求
 * 4、得到结果
 */
@Slf4j
public class RpcConsumerInvocation implements InvocationHandler {


    //需要一个注册中心，和一个接口类
    private Registry registry;

    private Class interfaceRef;

    private String group;

    public RpcConsumerInvocation(Registry registry, Class interfaceRef,String group) {
        this.registry = registry;
        this.interfaceRef = interfaceRef;
        this.group=group;
    }

    /**
     * 所有的方法调用都会走到这里
     *
     * @param proxy
     * @param method
     * @param args
     * @return
     * @throws Throwable
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        // 从接口中获取判断是否需要重试
        TryTimes annotation = method.getAnnotation(TryTimes.class);
        // 默认值0，代表不重试
        int tryTimes = 0;
        int intervalTime = 0;
        if (annotation != null) {
            tryTimes = annotation.tryTimes();
            intervalTime = annotation.intervalTime();
        }

        while (true) {
            // 什么情况下需要重试 1、异常 2、响应有问题 code =500

            //1.封装报文
            RequestPayload requestPayload = RequestPayload.builder()
                    .interfaceName(interfaceRef.getName())
                    .methodName(method.getName())
                    .parametersType(method.getParameterTypes())
                    .parametersValue(args)
                    .returnType(method.getReturnType())
                    .build();

            // todo 需要对各种请求id和各种类型做处理
            YrpcRequest yrpcRequest = YrpcRequest.builder()
                    .requestId(YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                    .compressType(CompressorFactory.getCompressor(YrpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                    .requestType(RequestType.REQUEST.getId())
                    .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                    .timeStamp(new Date().getTime())
                    .requestPayload(requestPayload)
                    .build();

            // 2.将请求存入本地线程，需要到合适的时候调用remove方法
            YrpcBootstrap.REQUEST_THREAD_LOCAL.set(yrpcRequest);

            // 3、发现服务，从注册中心拉取服务列表，通过客户端负载均衡寻找一个可用的服务
            // 传入服务的名字,返回ip+端口
            // 1.获取当前配置的负载均衡器，选取一个可用节点
            InetSocketAddress address = YrpcBootstrap.getInstance().getConfiguration()
                    .getLoadBalancer().selectServerAddress(interfaceRef.getName(),group);
            if (log.isDebugEnabled()) {
                log.debug("服务调用方，发现了服务【{}】的可用主机【{}】.",
                        interfaceRef.getName(), address);
            }

            //4.获取当前地址所队以对应的断路器， 如果断路器是打开的则不发送请求，抛出异常
            Map<SocketAddress, CircuitBreaker> everyIpCircuitBreaker = YrpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker();
            CircuitBreaker circuitBreaker = everyIpCircuitBreaker.get(address);
            if (circuitBreaker == null) {
                circuitBreaker = new CircuitBreaker(10, 0.5F);
                everyIpCircuitBreaker.put(address, circuitBreaker);
            }
            try {
                // 如果断路器是打开的
                if (yrpcRequest.getRequestType()!= RequestType.HEART_BEAT.getId()&&circuitBreaker.isBreak()) {
                    // 断路器打开，说明当前的请求不应该在发送
                    Timer timer = new Timer();
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            YrpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker().get(address).reset();
                        }
                    }, 5000);

                    throw new RuntimeException("当前断路器已经开启，无法发送请求");
                }

                // 使用netty连接服务器，发送 调用的 服务的名字+方法名字+参数列表，得到结果
                // 定义线程池，EventLoopGroup
                // q：整个连接过程放在这里行不行，也就意味着每次调用都会产生一个新的netty连接(对长连接不友好)。如何缓存我们的连接
                // 也就意味着，每次在此处建立一个新的连接是不合适的

                // 解决方案？缓存channel，尝试从缓存中获取channel，如果未获取，则创建新的连接，并进行缓存
                // 5、尝试从全局的缓存中获取一个通道
                Channel channel = getAvailableChannel(address);
                if (log.isDebugEnabled()) {
                    log.debug("、获取了和【{}】建立的连接通道，准备发送数据", address);
                }

                /**
                 * -----------------封装报文--------------
                 */


                /**
                 * -----------------同步策略--------------
                 */
//                ChannelFuture channelFuture = channel.writeAndFlush(new Object()).await();
//
//                if(channelFuture.isDone()){
//                    Object object = channelFuture.getNow();
//                }else{
//                    //异步执行异常
//                    Throwable cause = channelFuture.cause();
//                    throw new RuntimeException(cause);
//                }

                /**
                 * -----------------异步策略--------------
                 */

                //6.要写出报文
                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                //暴将露出去
                YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(), completableFuture);

                //这里writeAndFlush直接写出一个请求，这个请求的实例就会进入pipeLine，pipeline执行出站的一系列操作
                //我们可以想象得到，第一个出站程序一定是将 yrpcRequest --> 二进制的报文
                channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                    //我没想要的是服务端给我们的返回值
//                        if(promise.isDone()){
//                            completableFuture.complete(promise.getNow());
//                        }else
                    //处理异常即可
                    if (!promise.isSuccess()) {
                        completableFuture.completeExceptionally(promise.cause());
                    }
                }).await();

                //7. 清理ThreadLocal
                YrpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                //如果没有地方处理这个completableFuture，这里会发生阻塞，等待complete方法执行
                //q:我们需要在哪执行complete方法，很明显pipeline中最后的handler进行执行
                //8.获得响应的结果
                Object result = completableFuture.get(10, TimeUnit.SECONDS);

                return result;
            } catch (Exception e) {
                // 固定时间会有问题，产生重试风暴
                tryTimes--;
                // 记录错误的次数
                circuitBreaker.recordErrorRequest();
                try {
                    Thread.sleep(intervalTime);
                } catch (InterruptedException ex) {
                    log.error("在进行重试时发生异常", ex);
                }
                if (tryTimes < 0) {
                    log.error("对方法【{}】进行远程调用时，重试{}次，依然不可调用", method.getName(), tryTimes, e);
                    break;
                }
                log.error("在进行第{}次重试时发生异常", 3 - tryTimes, e);
            }
        }
        throw new RuntimeException("执行远程方法" + method.getName() + "调用失败");
    }

    /**
     * 根据地址获取可用通道
     *
     * @param address
     * @return
     */
    private Channel getAvailableChannel(InetSocketAddress address) {
        //1.尝试从缓存中获取
        Channel channel = YrpcBootstrap.CHANNEL_CACHE.get(address);

        //2.获取不到，就去建立连接
        if (channel == null) {
            // await 方法会阻塞，会等待连接成功在返回，netty还提供了异步处理的逻辑
//                    channel = NettyBootstrapInitializer.getBootstrap()
//                            .connect(address).await().channel();

            //使用addListener执行的异步操作
            CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener((ChannelFutureListener) promise -> {
                if (promise.isDone()) {
                    if (log.isDebugEnabled()) {
                        log.debug("已经和【{}】成功建立了连接", address);
                    }
                    channelFuture.complete(promise.channel());
                } else if (!promise.isSuccess()) {
                    channelFuture.completeExceptionally(promise.cause());
                }
            }).channel();

            //get是阻塞方法
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | TimeoutException | ExecutionException e) {
                log.error("获取通道时，发生异常。", e);
                throw new DiscoveryException(e);
            }

            // 缓存channel
            YrpcBootstrap.CHANNEL_CACHE.put(address, channel);
        }

        if (channel == null) {
            throw new NetworkException("获取通道是发生了异常。");
        }
        return channel;
    }
}
